package com.atguigu.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.util.DimUtil;
import com.atguigu.util.HBaseUtil;
import com.atguigu.util.JedisUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

public abstract class DimInfoRichMapFunction<T> extends RichMapFunction<T, T> {

    private Connection connection;
    private Jedis jedis;
    private String tableName;
    //sinkExtend:"00|,01|,02|" list:[00_,01_,02_,02|]
    //private List<String> list;

    public DimInfoRichMapFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = HBaseUtil.getConnection();
        jedis = JedisUtil.getJedis();

//        java.sql.Connection mysqlConn = DriverManager.getConnection(Constant.MYSQL_URL, "root", "000000");
//        List<JSONObject> queryList = JdbcUtil.queryList(mysqlConn, "select sink_extend from `gmall-230524-config`.table_process where sink_type = 'dim' and sink_table = '" + tableName + "' ", JSONObject.class, true);
//        JSONObject jsonObject = queryList.get(0);
//        String sinkExtend = jsonObject.getString("sink_extend");
//        if (sinkExtend != null) {
//            String[] split = sinkExtend.split(",");
//            list = new ArrayList<>();
//            for (String s : split) {
//                list.add(s.replace("|", "_"));
//            }
//            list.add(split[split.length - 1]);
//        }
//        mysqlConn.close;

//        Timer timer = new Timer();
//        timer.schedule(new TimerTask() {
//            @Override
//            public void run() {
//                aa();
//            }
//        }, 5000L, 10000L);
    }

    public abstract String getPk(T value);

    public abstract void join(T value, JSONObject dimInfo);

    @Override
    public T map(T value) throws Exception {

//        Class<?> valueClass = value.getClass();
//        Field skuId = valueClass.getField("skuId");
//        Object id = skuId.get(value);

        String pk = getPk(value);

//        if (list != null) {
//            pk = list.get(Math.abs(pk.hashCode() % list.size())) + pk;
//        }

        //查询维度信息
        JSONObject dimInfo = DimUtil.getDimInfo(connection, jedis, tableName, pk);

        //向T对象设置维度信息
        join(value, dimInfo);

        return value;
    }


    @Override
    public void close() throws Exception {
        connection.close();
        jedis.close();
    }
}
